package com.atguigu.flink.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * @author WEIYUNHUI
 * @date 2023/6/10 10:38
 * <p>
 * 批处理 wordcount => DataSet API
 * <p>
 * 注意: DataSet已经不推荐使用了。 此处， 只做基本演示。 方便和DataStream API 进行对比.
 * <p>
 * 编程套路:
 * 1. 创建执行环境
 * 2. 读取数据
 * 3. 对数据进行转换处理
 * 4. 输出结果
 */
public class Flink01_BatchWordCount {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        System.out.println(env);  //Local Environment (parallelism = 32).

        // 2. 读取数据
        // DataSource => DataSet
        DataSource<String> dataSource = env.readTextFile("input/words.txt");

        // 3. 对数据进行转换处理
        // 3.1 切分单词， 处理成(word,1)格式
        /*
          FlatMapFunction<T , O> :
             T: 输入类型
             O: 输出类型

             void flatMap(T value, Collector<O> out) throws Exception：
                 value : 输入的数据
                 out : 输出收集器， 用于将处理好的数据收集输出.
         */
        FlatMapOperator<String, Tuple2<String, Integer>> flatMapDs = dataSource.flatMap(
                //使用匿名内部类的方式来实现FlatMapFunction
                new FlatMapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {
                        String[] words = line.split(" ");
                        for (String word : words) {
                            out.collect(Tuple2.of(word, 1));
                        }
                    }
                }
        );

        // 3.2 按照单词分组
        /*
           groupBy(KeySelector<T, K> keyExtractor): 实现 KeySelector中的KEY getKey(IN value)方法， 方法的返回值就是分组的key.
           groupBy(int... fields) : Tuple中的第几个元素作为分组的key
           groupBy(String... fields): 一般数据是POJO(JavaBean)的时候，需要指定使用POJO的哪个属性作为分组的key
         */
        // 使用Tuple中的第一个元素进行分组
        UnsortedGrouping<Tuple2<String, Integer>> groupBy = flatMapDs.groupBy(0);

        // 3.3 汇总每个单词出现的次数
        //使用Tuple中的第二个元素进行汇总。
        AggregateOperator<Tuple2<String, Integer>> sumDs = groupBy.sum(1);

        // 4. 输出结果
        sumDs.print();

    }
}













